package com.shujia.spark.opt

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object Demo4MapPartition {
  def main(args: Array[String]): Unit = {
    //创建spark sql环境（新版spark统一的入口）
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("cache")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    //获取sparkContext
    val sc: SparkContext = spark.sparkContext

    //读取数据
    val studentRDD: RDD[String] = sc.textFile("data/students.txt")

    val kvRDD: RDD[(String, Int)] = studentRDD.mapPartitions((iter: Iterator[String]) => {

      /**
       * mapPartitions:一次处理一个分区的数据
       * 可以在处理分区数据之前连接外部的数据，减少数据库连接
       */
      iter.map(lint => {
        val clazz: String = lint.split(",")(4)
        (clazz, 1)
      })

    })

    kvRDD.foreach(println)

  }

}
